package com.kyhd.audio.config;

import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisSentinelConfiguration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.GenericToStringSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.kyhd.audio.service.RedisReceiver;

import lombok.extern.slf4j.Slf4j;


@Slf4j
@Configuration
@Import(PropertyConfig.class)
@ComponentScan("com.kyhd.audio.redis")
public class RedisHaConfiguration {
	
	@Value("${redis.sentinel.nodes}")
    private String sentinelNodes;
	
	@Value("${redis.database}")
	private String database;
	
	int timeout = 100;
	
	
	@Bean
	JedisConnectionFactory jedisConnectionFactory() {
		RedisSentinelConfiguration sentinelConfiguration = new RedisSentinelConfiguration();
		
		log.info("sentinelnodes:{}",sentinelNodes);
		List<String> nodeList = Lists.newArrayList(sentinelNodes.split(","));
		if(nodeList.size()>=2){
			String master = nodeList.remove(0);
			Set<String> sentinelNodes = Sets.newHashSet(nodeList);
			log.info("master:{},sentinelNodes:{}",master,sentinelNodes);
			sentinelConfiguration = new RedisSentinelConfiguration(master,sentinelNodes);
			
		}else{
			throw new RuntimeException("redis.sentinel.nodes: the num of nodes great than eqauls two at least.");
		}
		
	    JedisConnectionFactory jedisConFactory = new JedisConnectionFactory(sentinelConfiguration);
	    jedisConFactory.setTimeout(timeout);
	    log.info("host:{},port:{}",jedisConFactory.getHostName(),jedisConFactory.getPort());
	    jedisConFactory.setDatabase(Integer.valueOf(database));
	    return jedisConFactory;
	}
	
	
	@Bean(name="redisTemplate")
	public RedisTemplate<String, Object> redisTemplate() {
	    RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
	    template.setConnectionFactory(jedisConnectionFactory());
	    
	    template.setKeySerializer(new StringRedisSerializer());
	    template.setValueSerializer(new GenericToStringSerializer<Object>(Object.class));
	    return template;
	}
	
	
	@Bean
	public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
	    RedisMessageListenerContainer container = new RedisMessageListenerContainer();
	    container.setConnectionFactory(connectionFactory);
	    List<Topic> topics = Lists.newArrayList();
	    topics.add(new PatternTopic("novelstart"));
	    topics.add(new PatternTopic("novelhand"));
//	    container.addMessageListener(listenerAdapter, new PatternTopic("topic"));
	    container.addMessageListener(listenerAdapter, topics);
	    return container;
	}

	@Bean
	MessageListenerAdapter listenerAdapter(RedisReceiver receiver) {
	    //指定RedisReceiver为代理接收类,接收消息方法为receiveMessage
	    return new MessageListenerAdapter(receiver, "receiveMessage");
	}

   @Bean
    CountDownLatch latch() {
        return new CountDownLatch(1);
    }
	
	@Bean
	RedisReceiver receiver(CountDownLatch latch) {
	    return new RedisReceiver(latch);
	}
}
